package com.jblv.frame.common.kafka;

import com.jblv.frame.common.kafka.config.KafkaConfigProperties;
import kafka.producer.Partitioner;
import kafka.utils.VerifiableProperties;

/**
 * @author happyonion.yangtao
 * @version 1.0
 * @title DefaultPartitioner
 * @projectName jblv-frame
 * @packageName com.jblv.web.kafka
 * @description 自定义分区器
 * @date Create at 2020/4/17 14:18
 * @since jdk1.8
 */
public class DefaultPartitioner implements Partitioner {
    public DefaultPartitioner(VerifiableProperties props) {
    }

    @Override
    public int partition(Object key, int i) {
        System.out.println(key + ", " + i);
        String stringKey = (String)key;
        String[] strs = stringKey.split("\\.");
        int offset = Integer.valueOf(strs[1]);
        String value = KafkaConfigProperties.getValue(strs[0]);
        if (value != null && value != "") {
            int partition = offset % i;
            System.out.println(partition);
            return partition;
        } else {
            return 0;
        }
    }
}